package com.allen.flink.batch.join;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;

import java.util.ArrayList;

/**
 * 功能:默认Join
 *
 * @date: 2020-03-24 13:11
 * @author: Allen
 * @version: 0.0.4-snapshot
 * @Email: allenZyhang@163.com
 * @since: JDK 1.8
 **/
public class DefaultJoin {
    
    public static void main(String[] args) throws Exception {
        //获取执行环境
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        
        //第一个数据集Tuple 用户Id 用户名
        ArrayList<Tuple2<Integer, String>> list1 = new ArrayList<>();
        list1.add(new Tuple2<>(1, "Mars"));
        list1.add(new Tuple2<>(2, "Tom"));
        list1.add(new Tuple2<>(3, "Alan"));
        list1.add(new Tuple2<>(4, "Alex"));
        
        //第二个数据集Tuple 用户Id 用户城市
        ArrayList<Tuple2<Integer, String>> list2 = new ArrayList<>();
        list2.add(new Tuple2<>(1, "BeiJing"));
        list2.add(new Tuple2<>(2, "ShangHai"));
        list2.add(new Tuple2<>(3, "ChengDu"));
        
        DataSet<Tuple2<Integer, String>> ds1 = env.fromCollection(list1);
        DataSet<Tuple2<Integer, String>> ds2 = env.fromCollection(list2);
        DataSet<Tuple2<Tuple2<Integer, String>, Tuple2<Integer, String>>> joinData = ds1.join(ds2).where(0).equalTo(0);
        
        joinData.print();
    }
}
